package com.mjk.common.tools.scheduled;

import com.mjk.common.tools.logger.BizLogger;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.config.TriggerTask;
import org.springframework.scheduling.support.CronTrigger;

import java.util.List;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;


/**
 * 动态定时任务类
 * @author mjk
 *
 */
public abstract class DynamicScheduledHandler implements SchedulingConfigurer {
	private DynamicScheduledContent dsContainer;

    private final String CLASS_NAME = getClass().getSimpleName();

    /**
     * 获取用于执行任务的线程池
     */
    protected  ExecutorService getWorkerExecutor() {
    	return Executors.newScheduledThreadPool(20); // 20个线程来处理。
    }

    /**
     * 获取所有的任务信息
     */
    protected abstract List<DynamicScheduledModel> listTaskInfo();

    protected abstract void doProcess(DynamicScheduledModel model) ;

    @Override
    public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        dsContainer = new DynamicScheduledContent(taskRegistrar);
        // 每隔 100ms 调度一次，用于读取所有任务
        taskRegistrar.addFixedDelayTask(this::scheduleTask, 1000);
    }

    /**
     * 调度任务，加载所有任务并注册
     */
    private void scheduleTask() {
       listTaskInfo().forEach(taskInfo ->
                dsContainer.checkTask(taskInfo, new TriggerTask(() -> 
                        this.execute(taskInfo), triggerContext -> Objects.requireNonNull(new CronTrigger(taskInfo.getCron()).nextExecutionTime(triggerContext)).toInstant()
                ))
        );
    }

    private void execute(final DynamicScheduledModel taskInfo) {
        final String taskId = taskInfo.getId();
        try {
            Semaphore semaphore = dsContainer.getSemaphore(taskId);
            if (Objects.isNull(semaphore)) {
                BizLogger.error(CLASS_NAME+" semaphore is null, taskId: "+taskId);
                return;
            }
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    getWorkerExecutor().execute(() -> doProcess(taskInfo));
                } finally {
                    semaphore.release();
                }
            } else {
            	BizLogger.warn(CLASS_NAME+" too many executor, taskId: "+taskId);
            }
        } catch (InterruptedException e) {
        	BizLogger.warn(CLASS_NAME+"interruptedException error, taskId: "+taskId);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
        	BizLogger.error(CLASS_NAME+" execute error, taskId: "+taskId);
        }
    }
}
